import io
import pickle
import logging
import asyncio
import traceback
from minio import Minio
from typing import Any, BinaryIO, Iterator, Iterable
from minio.deleteobjects import DeleteObject
from utils.settings import MINIO_SETTING
from utils.util import Util

logger = Util.pyfile_logger(__file__)

def error_logs(func):
    def inner(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except:
            print(traceback.format_exc())
            logging.critical(traceback.format_exc())
            return False
    return inner


class BaseMinioUtil:
    def __init__(self, endpoint, access_key=None, secret_key=None, secure=False):
        # secure = True（默认推荐）：客户端通过HTTPS加密协议与服务器通信，数据在传输过程中被加密，安全性高。
        # secure = False：客户端通过HTTP明文协议与服务器通信，数据以未加密形式传输，安全性低。
        self.endpoint = endpoint
        self.access_key = access_key
        self.secret_key = secret_key
        self.secure = secure

    def _create_client(self):
        """
        确保线程安全: 每次调用创建新的 Minio 客户端实例
        """
        return Minio(
            endpoint=self.endpoint,
            access_key=self.access_key,
            secret_key=self.secret_key,
            secure=self.secure
        )

    async def make_bucket(self, bucket_name: str):
        """
        创建桶
        :param bucket_name: 桶名称
        :return: bool
        """

        def _make_bucket():
            mc = self._create_client()
            try:
                if not mc.bucket_exists(bucket_name=bucket_name):
                    mc.make_bucket(bucket_name=bucket_name)
                    return True
            except:
                print(traceback.format_exc())
            return False
        return await asyncio.to_thread(_make_bucket)

    @error_logs
    async def list_buckets(self):
        """获取桶列表"""

        def _list_buckets():
            mc = self._create_client()
            return mc.list_buckets()
        return await asyncio.to_thread(_list_buckets)

    @error_logs
    async def remove_bucket(self, bucket_name: str):
        """删除 bucket"""

        def _remove_bucket():
            mc = self._create_client()
            return mc.remove_bucket(bucket_name)
        return await asyncio.to_thread(_remove_bucket)

    @error_logs
    async def put_object(self, bucket_name: str, object_name: str, data: BinaryIO, length: int):
        """将数据流上传到 bucket 中的对象"""

        def _put_object():
            mc = self._create_client()
            # self.make_bucket(bucket_name)
            return mc.put_object(bucket_name, object_name, data, length)
        return await asyncio.to_thread(_put_object)

    @error_logs
    async def fput_object(self, bucket_name: str, object_name: str, file_path: str,):
        """将文件上传到 bucket 中的对象"""

        def _put_object():
            mc = self._create_client()
            self.make_bucket(bucket_name)
            return mc.fput_object(bucket_name, object_name, file_path)
        return await asyncio.to_thread(_put_object)

    @error_logs
    async def get_object(self, bucket_name: str, object_name: str):
        """获取 bucket 中的流数据"""

        def _get_object():
            mc = self._create_client()
            return mc.get_object(bucket_name, object_name)
        return await asyncio.to_thread(_get_object)

    @error_logs
    async def fget_object(self, bucket_name: str, object_name: str, file_path: str):
        """获取 bucket 中的文件数据"""

        def _fget_object():
            mc = self._create_client()
            return mc.fget_object(bucket_name, object_name, file_path)
        return await asyncio.to_thread(_fget_object)

    @error_logs
    async def list_objects(self, bucket_name: str, prefix=None, start_after=None, recursive=True):
        """
        获取 bucket 中的数据列表
        recursive=True: 当你想获取某个前缀下所有层级的对象时，适合用于需要全面检索的场景。
        recursive=False: 当你只想获取当前层级下的对象时，适合用于更精确的对象查找。
        """

        def _list_objects():
            mc = self._create_client()
            return mc.list_objects(bucket_name, prefix=prefix, start_after=start_after, recursive=recursive)
        return await asyncio.to_thread(_list_objects)

    @error_logs
    async def remove_object(self, bucket_name: str, object_name: str):
        """删除数据对象"""

        def _remove_object():
            mc = self._create_client()
            return mc.remove_object(bucket_name, object_name)
        return await asyncio.to_thread(_remove_object)

    @error_logs
    async def remove_objects(self, bucket_name: str, delete_object_list: Iterable[DeleteObject]):
        """批量删除数据对象"""

        def _remove_objects():
            delete_files = [DeleteObject(file) for file in delete_object_list]
            mc = self._create_client()
            for error in mc.remove_objects(bucket_name, delete_files):
                print("error", error)
        return await asyncio.to_thread(_remove_objects)


class AsyncMinioUtil(BaseMinioUtil):
    def __init__(self,
                 endpoint=MINIO_SETTING['endpoint'],
                 access_key=MINIO_SETTING['access_key'],
                 secret_key=MINIO_SETTING['secret_key'],
                 secure=MINIO_SETTING['secure']
                 ):
        super().__init__(endpoint, access_key=access_key, secret_key=secret_key, secure=secure)

    def check_data_format(self, data):
        """检查数据格式是否符合"""
        keys = ['equipment_uid', 'parameter_uid', 'timestamp', 'value']
        for key in keys:
            if key not in data.keys():
                raise Exception(f"缺少必要字段：{keys}")
        if not isinstance(data['value'], list):
            raise Exception("值类型必须是列表")

    def get_object_unique_name(self, parameter_uid, collecting_time):
        time_str = collecting_time.strftime('%Y%m%d%H%M%S')
        return f"{parameter_uid}/{time_str}.pkl"

    def get_bucket_name(self, equipment_uid: str):
        """格式化 bucket 名称"""
        return f"equipment-{equipment_uid.replace('_', '-')}"

    async def create_bucket_for_equipments(self, ids):
        """批量创建 bucket"""
        for id in ids:
            await self.make_bucket(self.get_bucket_name(id))

    async def insert_wave_data(self, data: dict):
        """
        将波形数据存入 minio
        :param data: {
            "timestamp": 1742126423000,  # 毫秒
            "type": "list",
            "equipment_uid": "2_2#CAS_GW_AC2",
            "parameter_uid": "vib_CF_H_A_128K_20K",
            "value": [1.3176, -0.58194, -0.58194, -0.63684, ...]
        }
        """
        self.check_data_format(data)

        # 判断时间戳是秒或毫秒
        len_timestamp = len(str(data.get('timestamp')))
        if len_timestamp == 10:
            collecting_time = Util.timestamp2datetime(data['timestamp'])
        elif len_timestamp == 13:
            collecting_time = Util.timestamp2datetime(int(data['timestamp'] / 1000))
        else:
            raise Exception("timestamp only must be second or microsecond!")

        # 生成 bucket_name、object_name
        bucket_name = self.get_bucket_name(data.get('equipment_uid'))
        object_name = self.get_object_unique_name(data.get('parameter_uid'), collecting_time)

        # 将字节流波形发送到 minio
        value_bytes = pickle.dumps(data.get('value'))
        await self.make_bucket(bucket_name)
        ret = await self.put_object(bucket_name, object_name, io.BytesIO(value_bytes), len(value_bytes))
        if not ret:
            logger.error('发送数据失败')
            return 'failure'
